package com.study.config;

import com.redis.queue.AdaptiveListener;
import com.redis.queue.CacheConstant;
import com.redis.queue.RedisQueueUtils;
import com.redis.queue.StreamErrorHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Configuration
@Slf4j
public class RedisStreamConfiguration {

    @Autowired
    private RedisQueueUtils redisQueueUtils;

    @Autowired
    TaskExecutor taskExecutorStream;

    @Autowired
    private RedisConnectionFactory redisConnectionFactory;

    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer<String, ObjectRecord<String, String>> streamMessageListenerContainer(List<AdaptiveListener> adaptiveListeners) {
        AtomicInteger index = new AtomicInteger(1);
        int processors = Runtime.getRuntime().availableProcessors();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(), r -> {
            Thread thread = new Thread(r);
            thread.setName("redis async-stream-consumer-" + index.getAndIncrement());
            thread.setDaemon(true);
            return thread;
        });

        redisQueueUtils.createGroup(CacheConstant.MESSAGE_STEAM,CacheConstant.MESSAGE_GROUP);
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String,ObjectRecord<String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        // 一次最多获取多少条消息
                        .batchSize(1)
                        // 运行 Stream 的 poll task
                        .executor(executor)
                        .targetType(String.class)
                        // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小
                        .pollTimeout(Duration.ofSeconds(30))
                        // 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理
                        .errorHandler(new StreamErrorHandler())
                        .build();


        StreamMessageListenerContainer<String, ObjectRecord<String, String>> streamMessageListenerContainer =  StreamMessageListenerContainer.create(redisConnectionFactory, options);
        // 消费组A,不自动ack
        /**
         * 设置消费者名称和消费者所在的消费组名称
         */
        adaptiveListeners.forEach(adaptiveListener -> {
            adaptiveListener.setRedisQueueUtils(redisQueueUtils);
            streamMessageListenerContainer.receiveAutoAck(Consumer.from(CacheConstant.MESSAGE_GROUP, CacheConstant.MESSAGE_CONSUMER),StreamOffset.create(CacheConstant.MESSAGE_STEAM,ReadOffset.lastConsumed()),adaptiveListener);
        });
        return streamMessageListenerContainer;
    }
}
